package org.apache.ocean;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.document.Document;
import org.apache.lucene.index.CorruptIndexException;
import org.apache.lucene.index.LogDocMergePolicy;
import org.apache.lucene.index.Term;
import org.apache.lucene.store.IndexInput;
import org.apache.lucene.store.RAMDirectory;

/**
 * 
 * 
 * @author jasonr
 */
// TODO: unit test RAMIndexManager
// TODO: handle the case of being closed
public class RamIndexManager {
	public static Logger log = Logger.getLogger(RamIndexManager.class.getName());
	// indices that no longer have documents added to them
	private LinkedBlockingQueue<RamIndex> finishedRAMIndices = new LinkedBlockingQueue<RamIndex>();
	// a lock on changing any of the ram indices. read lock for using, write lock for changing
	private ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
	private CDouble maxIndexSizeMB;
	private Analyzer analyzer;
	private RamIndex currentRAMIndex;
	private boolean isClosed = false;
  
	public IndexInput getIndexFileInput(String indexID, String snapshotName, String fileName) throws IOException {
		RamIndex ramIndex = getIndex(indexID);
		if (ramIndex != null) {
			return ramIndex.getIndexFileInput(snapshotName, fileName);
		} else {
			return null;
		}
	}
	
	public RamIndex getIndex(String indexID) {
		readLock();
		try {
			for (RamIndex ramIndex : finishedRAMIndices) {
				if (ramIndex.getID().equals(indexID)) return ramIndex;
			}
			if (currentRAMIndex != null) {
				if (currentRAMIndex.getID().equals(indexID)) return currentRAMIndex;
			}
			return null;
		} finally {
			readUnlock();
		}
	}
	
	public RamIndexManager(CDouble maxIndexSizeMB, Analyzer analyzer) {
		this.maxIndexSizeMB = maxIndexSizeMB;
		this.analyzer = analyzer;
	}
  
	public Map<String,ShardInfo.Index> createShardInfoIndexes() throws Exception {
		Map<String,ShardInfo.Index> map = new HashMap<String,ShardInfo.Index>();
		for (RamIndex ramIndex : finishedRAMIndices) {
			ShardInfo.Index shardInfoIndex = new ShardInfo.Index();
			shardInfoIndex.id = ramIndex.getID();
			shardInfoIndex.segmentsVersion = ramIndex.getSegmentsVersion();
			shardInfoIndex.type = ShardInfo.Index.RAM;
			shardInfoIndex.status = ShardInfo.Index.FINISHED;
			shardInfoIndex.snapshot = ramIndex.getLatestSnapshot();
			//shardInfoIndex.addAll(ClusterUtil.createIndexFiles(ramIndex.getDirectory()));
			map.put(shardInfoIndex.id, shardInfoIndex); 
		}
		if (currentRAMIndex != null) {
			readLock();
			try {
				ShardInfo.Index shardInfoIndex = new ShardInfo.Index();
				shardInfoIndex.id = currentRAMIndex.getID();
				shardInfoIndex.segmentsVersion = currentRAMIndex.getSegmentsVersion();
				shardInfoIndex.type = ShardInfo.Index.RAM;
				shardInfoIndex.status = ShardInfo.Index.CURRENT;
				shardInfoIndex.snapshot = currentRAMIndex.getLatestSnapshot();
				//shardInfoIndex.addAll(ClusterUtil.createIndexFiles(currentRAMIndex.getDirectory()));
				map.put(shardInfoIndex.id, shardInfoIndex);
			} finally {
				readUnlock();
			}
		}
		return map;
	}
	
	public void close() {
		lock.writeLock().lock();
		try {
			for (ClusterIndexModifier clusterIndexModifier : finishedRAMIndices) {
				try {
					clusterIndexModifier.close();
				} catch (Throwable throwable) {
					log.log(Level.SEVERE, "", throwable);
				}
			}
			if (currentRAMIndex != null) {
				try {
					currentRAMIndex.close();
				} catch (Throwable throwable) {
					log.log(Level.SEVERE, "", throwable);
				}
			}
		} finally {
			lock.writeLock().unlock();
		}
	}

	public List<Term> getTermPerIndex(Term startTerm) throws Exception {
		List<Term> terms = new ArrayList<Term>();
		for (RamIndex ramIndex : finishedRAMIndices) {
			Term term = ramIndex.getTerm(startTerm);
			if (term != null)
				terms.add(term);
		}
		if (currentRAMIndex != null) {
			Term term = currentRAMIndex.getTerm(startTerm);
			if (term != null)
				terms.add(term);
		}
		return terms;
	}

	public RamIndex getRAMIndexNewestNoLock() {
		if (currentRAMIndex != null)
			return currentRAMIndex;
		RamIndex lastRAMIndex = null;
		for (RamIndex ramIndex : finishedRAMIndices) {
			lastRAMIndex = ramIndex;
		}
		return lastRAMIndex;
	}

	public RamIndex getRAMIndexFinishedOldestNoLock() {
		return finishedRAMIndices.peek();
	}

	public List<RamIndex> getFinishedIndicesNoLock(int max) {
		Iterator<RamIndex> iterator = finishedRAMIndices.iterator();
		List<RamIndex> list = new ArrayList<RamIndex>();
		int count = 0;
		while (iterator.hasNext() && count < max) {
			list.add(iterator.next());
		}
		return list;
	}

	void removeRAMIndexFinished(List<RamIndex> ramIndices) {
		LockUtil.lock(lock.writeLock());
		try {
			for (RamIndex ramIndex : ramIndices) {
				finishedRAMIndices.remove(ramIndex);
			}
		} finally {
			lock.writeLock().unlock();
		}
	}

	void removeRAMIndexFinished(RamIndex ramIndex) {
		LockUtil.lock(lock.writeLock());
		try {
			finishedRAMIndices.remove(ramIndex);
		} finally {
			lock.writeLock().unlock();
		}
	}

	public void reload() throws CorruptIndexException, IOException, Exception {
		readLock();
		try {
			for (RamIndex ramIndex : finishedRAMIndices) {
				ramIndex.reload();
			}
			if (currentRAMIndex != null)
				currentRAMIndex.reload();
		} finally {
			readUnlock();
		}
	}

	public CLong deleteDocuments(Term term, boolean reloadBeforeDelete, boolean ifDeletedReload) throws Exception {
		readLock();
		try {
			CLong total = new CLong(0);
			for (RamIndex ramIndex : finishedRAMIndices) {
				if (reloadBeforeDelete) ramIndex.reload();
				CLong deleted = ramIndex.deleteDocuments(term, reloadBeforeDelete, true, ifDeletedReload);
				if (ifDeletedReload && deleted.value > 0)
					ramIndex.reload();
				total.value += deleted.value;
			}
			if (currentRAMIndex != null) {
				readLock();
				try {
					if (reloadBeforeDelete) currentRAMIndex.reload();
					CLong deleted = currentRAMIndex.deleteDocuments(term, reloadBeforeDelete, true, ifDeletedReload);
					if (ifDeletedReload && deleted.value > 0)
						currentRAMIndex.reload();
					total.value += deleted.value;
				} finally {
					readUnlock();
				}
			}
			return total;
		} finally {
			readUnlock();
		}
	}

	public void readLock() {
		lock.readLock().lock();
	}

	public void readUnlock() {
		lock.readLock().unlock();
	}

	public CDouble getSizeMB() throws IOException {
		CDouble total = new CDouble(0.0);
		readLock();
		try {
			for (RamIndex ramIndex : finishedRAMIndices) {
				total.value += ramIndex.getSizeMB().value;
			}
			return total;
		} finally {
			readUnlock();
		}
	}

	public List<SegmentIndexSearcher> createSegmentIndexSearchers() throws Exception {
		readLock();
		try {
			List<SegmentIndexSearcher> indexSearchers = new ArrayList<SegmentIndexSearcher>(finishedRAMIndices.size() + 1);
			for (ClusterIndexModifier ramIndex : finishedRAMIndices) {
				indexSearchers.addAll(ramIndex.createSegmentIndexSearchers());
			}
			if (currentRAMIndex != null)
				indexSearchers.addAll(currentRAMIndex.createSegmentIndexSearchers());
			return indexSearchers;
		} finally {
			readUnlock();
		}
	}

	public List<MultiIndexSearcher> createMultiIndexSearchers() throws Exception {
		readLock();
		try {
			List<MultiIndexSearcher> indexSearchers = new ArrayList<MultiIndexSearcher>(finishedRAMIndices.size() + 1);
			for (ClusterIndexModifier ramIndex : finishedRAMIndices) {
				indexSearchers.add(ramIndex.createMultiIndexSearcher());
			}
			if (currentRAMIndex != null)
				indexSearchers.add(currentRAMIndex.createMultiIndexSearcher());
			return indexSearchers;
		} finally {
			readUnlock();
		}
	}

	public void openRamIndex() throws Exception {
		if (currentRAMIndex == null) {
			LockUtil.lock(lock.writeLock());
			try {
				if (currentRAMIndex == null) {
					RAMDirectory directory = new RAMDirectory();
					LogDocMergePolicy mergePolicy = new LogDocMergePolicy();
					mergePolicy.setMinMergeDocs(10);
					mergePolicy.setMaxMergeDocs(1000);
					mergePolicy.setMergeFactor(100);
					currentRAMIndex = new RamIndex(new CDouble(1.0), mergePolicy, directory, analyzer);
				}
			} finally {
				lock.writeLock().unlock();
			}
		}
	}

	public void flush() throws Exception {
		LockUtil.lock(lock.writeLock());
		try {
			if (currentRAMIndex != null) {
				finishedRAMIndices.add(currentRAMIndex);
				currentRAMIndex = null;
			}
		} finally {
			lock.writeLock().unlock();
		}
	}

	private void checkSize() throws Exception {
		if (currentRAMIndex != null) {
			if (currentRAMIndex.getSizeMB().value > maxIndexSizeMB.value) {
				flush();
			}
		}
	}
  
	public void addDocuments(List<Document> documents, Analyzer analyzer) throws Exception {
		openRamIndex();
		readLock();
		try {
			currentRAMIndex.addDocuments(documents, analyzer, true);
		} finally {
			readUnlock();
		}
		checkSize();
	}
	
	public void addDocument(Document document, Analyzer analyzer) throws Exception {
		openRamIndex();
		readLock();
		try {
			currentRAMIndex.addDocument(document, analyzer, true);
		} finally {
			readUnlock();
		}
		checkSize();
	}
}
